package cn.kennylee.codehub.mq.lip.integration.kafka;

import cn.kennylee.codehub.mq.lip.integration.kafka.demo.HelloKafkaListener;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;

/**
 * <p> SpringBoot单元测试入口 </p>
 * <p>Created on 2024/1/27.</p>
 *
 * @author kennylee
 * @since 0.0.1
 */
@SpringBootApplication
public class MqTestApp {

    public static void main(String[] args) {
        SpringApplication.run(MqTestApp.class, args);
    }


    @Bean
    public ConcurrentMessageListenerContainer<String, String> kafkaContainer(
        ConsumerFactory<String, String> consumerFactory,
        HelloKafkaListener helloKafkaListener) {

        // 1. 配置容器属性（指定Topic和监听器）
        ContainerProperties containerProps = new ContainerProperties(helloKafkaListener.getTopic());
        containerProps.setGroupId(helloKafkaListener.getConsumerGroup()); // 消费组ID
        containerProps.setAckMode(ContainerProperties.AckMode.BATCH); // 批量确认模式
        containerProps.setMessageListener(helloKafkaListener); // 关键绑定

        // 2. 创建容器并设置并发
        ConcurrentMessageListenerContainer<String, String> container = new ConcurrentMessageListenerContainer<>(consumerFactory, containerProps);
        container.setConcurrency(3); // 并发消费者数量
        return container;
    }
}
